package service

import (
	"bytes"
	"google.golang.org/protobuf/proto"
	"src/config"
	connection2 "src/connection"
	"src/model/ProtoModel"
)

type Sub struct {
	Act         ProtoModel.SendActProto
	Type        ProtoModel.DataTypeProto
	Uid         string
	Sid         string
	Cid         int64
	RedisClient *connection2.PubRedis
	Buffer      bytes.Buffer
}

func (this *Sub) Packet() *Sub {
	buf := bytes.Buffer{}
	buf.WriteByte(byte(this.Act.Number()))
	buf.WriteByte(byte(this.Type.Number()))

	this.Buffer = buf
	//this.S.Unlock()
	return this
}

// 打包数据对推送服务器的redis 订阅格式
func (this *Sub) Message(mes *ProtoModel.MessagesPubProto) *Sub {
	d, err := proto.Marshal(mes)
	if err != nil {
		return nil
	}
	buf := bytes.Buffer{}
	buf.WriteByte(byte(this.Act.Number()))
	buf.WriteByte(byte(this.Type.Number()))
	buf.Write(d)
	this.Buffer = buf
	return this
}

// 打包数据对推送服务器的redis 订阅格式 同步
func (this *Sub) SendRedisDo() error {
	client := (&connection2.PubRedis{}).NewRedisClient()
	data := this.Buffer
	_, err := client.DoPublish("chat", data.Bytes())
	if err != nil {
		return err
	}
	return nil
}

// 打包数据对推送服务器的redis 订阅格式 异步
func (this *Sub) SendRedisPublish() error {
	channelName := config.V().GetString("subscribe.redis.channelName")
	client := &connection2.PubRedis{}
	data := this.Buffer.Bytes()
	err := client.SendPublish(channelName, data)
	//err := client.Send("Publish", "chat", data.Bytes())
	if err != nil {
		return err
	}
	return nil
}
func (this *Sub) SendKafka() error {
	topicName := config.V().GetString("subscribe.kafka.topicName")
	Kafka := (&connection2.PubKafkaConsumer{}).NewKafkaConsumer()
	data := this.Buffer
	err := Kafka.Push(topicName, "", data.Bytes())
	if err != nil {
		return err
	}
	return nil
}
